package com.jthinking.service.kafka.delay.queue;

import com.jthinking.service.kafka.delay.queue.config.TimerProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.LinkedList;

@Component
@ConditionalOnProperty(value = {"timer.enable"})
public class TimerRunner implements ApplicationRunner {

    private static final Logger LOGGER = LoggerFactory.getLogger(TimerRunner.class);

    @Resource
    private KafkaTemplate<String, Long> longKafkaTemplate;
    @Resource
    private TimerProperties timerProperties;

    private final LinkedList<Long> queue = new LinkedList<>();

    @Override
    public void run(ApplicationArguments args) throws Exception {

        // List<String> servers = timerProperties.getServers();

        Thread thread = new Thread(() -> {
            while (true) {
                Long currentTimeMillis = queue.poll();
                if (currentTimeMillis != null) {
                    longKafkaTemplate.send("delay-queue-timeout", currentTimeMillis);
                    continue;
                }
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    LOGGER.error("", e);
                }
            }
        });
        thread.setDaemon(true);
        thread.start();

        Calendar instance = Calendar.getInstance();
        instance.add(Calendar.SECOND, 1);
        instance.set(Calendar.MILLISECOND, 0);
        long lastTimeMillis = instance.getTimeInMillis();
        while (true) {
            long currentTimeMillis = System.currentTimeMillis();
            long duration = currentTimeMillis - lastTimeMillis;
            if (duration >= 1000) {
                lastTimeMillis = currentTimeMillis - duration + 1000;
                //LOGGER.info("Timeout! currentTimeMillis={}", currentTimeMillis);
                queue.add(currentTimeMillis);
            }
        }
    }

}